/**
 * 
 */
package edu.cmu.vlis.wassup.publisher;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Properties;

import javax.jms.JMSException;
import javax.jms.Session;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;

import edu.cmu.vlis.wassup.CoordinatorResponder;
import edu.cmu.vlis.wassup.IProcess;

import sun.misc.Signal;

/**
 * This is an abstract poller which all poller must inherit.
 */
public abstract class AbstractPoller implements Runnable, IProcess{

    TopicSession topicSession;
    TopicPublisher topicPublisher;
    TopicConnection topicConn;
    CoordinatorResponder cr;
    boolean stop;
    String name;

    public AbstractPoller(int id) {
        //register hook method
        Runtime.getRuntime().addShutdownHook(new
                ShutdownPoller(topicConn));
        loadProperties();
        TopicConnectionFactory topicConnFactory = new com.sun.messaging.TopicConnectionFactory();
        try {
            String jmsURL = InetAddress.getLocalHost().getHostAddress()+":7676"; // Queues will always be on localhost.
            ((com.sun.messaging.TopicConnectionFactory) topicConnFactory).setProperty(
                    com.sun.messaging.ConnectionConfiguration.imqAddressList, jmsURL);
            ((com.sun.messaging.TopicConnectionFactory) topicConnFactory).setProperty(
                    com.sun.messaging.ConnectionConfiguration.imqDefaultUsername, "admin");
            ((com.sun.messaging.TopicConnectionFactory) topicConnFactory).setProperty(
                    com.sun.messaging.ConnectionConfiguration.imqDefaultPassword, "admin");

            this.topicConn= topicConnFactory.createTopicConnection();
            this.topicSession = this.topicConn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
            this.topicPublisher = this.topicSession.createPublisher(null); 

        } catch (JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (UnknownHostException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        stop = false;
    }

    public void start(String name){

        Thread pollerThread = new Thread(this);
        pollerThread.start();
        this.name = name;        
        Thread crThread = new Thread(new CoordinatorResponder(getID()));
        crThread.start();
    }   
    
    public void run(){
        this.startPoller();
    }
    
    private void loadProperties(){
        Properties props = new Properties();
        try {
            props.load(new FileInputStream("wassup/src/properties"));
            System.getProperties().putAll(props);

        } catch (FileNotFoundException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    protected void  startPoller(){
        try {
            topicConn.start();

            stop = false;
            while(!stop) {  
                //Keep polling after 10 seconds
                try {
                    int msgPolled = getMessagesNputInPublisherQ();
                    System.out.printf("%s polled %d messages.", getPollerName(),msgPolled);
                    System.out.println();
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        } catch (JMSException e1) {
            // TODO Auto-generated catch block
            e1.printStackTrace();
        }
    }

    public void stop()  {
        stop = true;
        try {
            if (topicConn != null)
                topicConn.stop();
            System.out.println("Stoped the Poller");
        } catch (JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    public void shutDown() {
        System.out.println("Sending Int signal");
        Signal.raise(new Signal("INT"));
    }
    /**
     * Function which actually does the polling.
     * @return returns number of messages polled.
     */
    public abstract int getMessagesNputInPublisherQ();

    public abstract String getPollerName();

    public abstract int getID();

}
